package cn.com.wind.bdg.polish.polish.service;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import cn.com.wind.bdg.polish.polish.utils.GetHTMLData;
import cn.com.wind.bdg.polish.polish.utils.GetHtml;
import cn.com.wind.bdg.polish.polish.utils.JsoupGetCode;

/**
 * 
 * @ClassName:  UrlData2KafkaService   
 * @Description:发送详情数据到kafka
 * @author: gaoxiang
 * @date:   2020年8月19日 下午3:28:21
 */
@Service
public class UrlData2KafkaService {
	private static Logger LOGGER = LoggerFactory.getLogger(UrlData2KafkaService.class);
	@Autowired
    private KafkaService kafkaService;
	@Autowired
	GetHtml getHtml;
	public void dealKafka() throws Exception {
		List<String> resUrl = new ArrayList<String>();
		GetHTMLData getHTMLData = new GetHTMLData();
		resUrl = getHtml.method();
		List<List<String>> listRes = new ArrayList<List<String>>();
		JsoupGetCode jsoupGetCode = new JsoupGetCode();
		//先将每一组数据存入map，然后以公司名称为key，map为value，存入kafka
		kafkaService.init();
		for(String url:resUrl) {
			Map<String, String> mapData = new HashMap<String, String>();
			mapData.put("CompanyName", getHTMLData.getCompanyName(url));
			mapData.put("Type", getHTMLData.getType(url));
			mapData.put("Location", getHTMLData.getLocation(url));
			mapData.put("Area", getHTMLData.getArea(url));
			//如果code为空，则不执行如kafka操作，相关信息输出到日志，并跳到下一个循环
			if(jsoupGetCode.method(getHTMLData.getArea(url))==null) {
				LOGGER.info("地域:{}，所对应code无值", getHTMLData.getArea(url));
				continue;
			}else {
				mapData.put("Code", jsoupGetCode.method(getHTMLData.getArea(url)));
			}
			String resData = mapData.toString();
			//将key设置为公司名称，value设置为存储相关信息的map，存入kafka
			kafkaService.produce(getHTMLData.getCompanyName(url), resData);
			System.out.println("data存入kafka成功");
			
//			kafkaService.produce("CompanyName", getHTMLData.getCompanyName(url));
//			kafkaService.produce("Type", getHTMLData.getType(url));
//			kafkaService.produce("Location", getHTMLData.getLocation(url));
//			kafkaService.produce("Area", getHTMLData.getArea(url));
//			try {
//				kafkaService.produce("Code", jsoupGetCode.method(getHTMLData.getArea(url)));
//			} catch (Exception e) {
//				//校验不通过将错误信息存入error.log日志文件
//				LOGGER.error("地域:{}，所对应code无值", getHTMLData.getArea(url), e);
//			}	
//			System.out.println("data存入kafka成功");
		}	
	}
}
